package org.czy.kafka.test.comsumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

/**
 * KafkaProducerTest
 *
 * @author caizhiyang
 * @since 2024-04-25
 */
public class KafkaConsumerOffsetTest4 {

    public static void main(String[] args) {
        //1.配置属性集合
        Map<String, Object> configMap = new HashMap<>();

        //2.设置kafka集群地址
        configMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");

        //3.配置属性：Kafka生产的数据为KV对，所以在生产数据进行传输前需要分别对K,V进行对应的序列化操作
        configMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        //设置禁用自动提交偏移量
        configMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);

        //设置消费者的事务隔离级别,默认是read_committed
        configMap.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_uncommitted");


        //5.配置属性: 消费者组
        configMap.put("group.id", "atguigu5");

        //6.配置属性: 自动提交偏移量
        configMap.put("enable.auto.commit", "true");

        //7.创建消费者对象
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(configMap);

        //8.消费者对象订阅topic
        consumer.subscribe(Collections.singletonList("test"));


        //拉取数据
        boolean flag = true;
        while (flag) {
            consumer.poll(Duration.ofMillis(100));
            //获取集群信息
            Set<TopicPartition> assignment = consumer.assignment();
            for (TopicPartition topicPartition : assignment) {
                //匹配指定topic
                if ("test".equals(topicPartition.topic())) {
                    //手动设置offset
                    consumer.seek(topicPartition,2);
                    flag=false;
                }
            }
        }


        while (true) {
            //9.每隔100毫秒，抓取一次数据
            ConsumerRecords<String, String> records =
                    consumer.poll(Duration.ofMillis(100));
            //10.打印抓取的数据
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("K = " + record.key() + ", V = " + record.value());
            }


            //同步提交偏移量
//            consumer.commitSync();

            //异步提交偏移量
            consumer.commitAsync();
        }
    }
}
